package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/1/7 08:26
 */
public class Dws_02_DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_02_DwsTrafficVcChArIsNewPageViewWindow().init(
            4002,
            2,
            "Dws_02_DwsTrafficVcChArIsNewPageViewWindow",
            Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        // 1. 解析成 pojo 类型
        SingleOutputStreamOperator<TrafficPageViewBean> beanStream = parseToPojo(stream);
        // 2. 开窗聚和
        SingleOutputStreamOperator<TrafficPageViewBean> resultStream = windowAndAgg(beanStream);
//        resultStream.print();
//         3. 写出到 clickhouse 中
        writeToClickHouse(resultStream);
        
    }
    
    private void writeToClickHouse(SingleOutputStreamOperator<TrafficPageViewBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil
                                 .getClickHouseSink("dws_traffic_vc_ch_ar_is_new_page_view_window",
                                                    TrafficPageViewBean.class
                                 ));
    }
    
    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(
        SingleOutputStreamOperator<TrafficPageViewBean> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    // 当一个并行度的水印超过 60s 不更新, 则以其他的并行度的水印为准
                    .withIdleness(Duration.ofSeconds(60))  // 防止数据倾斜导致的水印不更新
            )
            .keyBy(bean -> bean.getVc() + "_" + bean.getCh() + "_" + bean.getAr() + "_" + bean.getIsNew())
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1,
                                                      TrafficPageViewBean value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        // 前面聚合函数最终的聚和的结果!
                                        // 有且仅有一个值
                                        Iterable<TrafficPageViewBean> elements,
                                        Collector<TrafficPageViewBean> out) throws Exception {
                        TrafficPageViewBean bean = elements.iterator().next();
                        
                        // 增加下窗口的开始时间和结束时间
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                        // 改成结果的统计时间: 用系统时间
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                        
                        
                    }
                }
            );
    }
    
    private SingleOutputStreamOperator<TrafficPageViewBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
            .map(JSON::parseObject)
            .keyBy(obj -> obj.getString("mid"))
            .process(new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                
                private ValueState<String> lastVisitDateState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDate", String.class));
                }
                
                @Override
                public void processElement(JSONObject obj,
                                           Context ctx,
                                           Collector<TrafficPageViewBean> out) throws Exception {
                    String vc = obj.getString("vc");
                    String ch = obj.getString("ch");
                    String ar = obj.getString("ar");
                    String isNew = obj.getString("is_new");
                    Long durSum = obj.getLong("during_time");
                    Long ts = obj.getLong("ts");
                    
                    String today = AtguiguUtil.tsToDate(ts);
                    
                    Long uvCt = 0L;  //
                    // 如何知道这条日志, 是这个设备当天的第一条日志
                    // 状态: 存储条设备的最后一次访问的日期(年月日)
                    // 这条数据过来, 计算出他的年月日, 如果和状态一样, 证明是同一天
                    // 2023-01-07   2023-01-08
                    if (!today.equals(lastVisitDateState.value())) {
                        // 今天的第一条数据来了
                        uvCt = 1L;
                        // 更新状态
                        lastVisitDateState.update(today);
                    }
                    Long svCt = 0L;  //
                    String lastPageId = obj.getString("last_page_id");
                    if (lastPageId == null) {
                        // 这条日志, 是用户新的会话
                        svCt = 1L;
                    }
                    
                    out.collect(new TrafficPageViewBean(
                        "", "",
                        vc, ch, ar, isNew,
                        uvCt, svCt, 1L, durSum,
                        ts
                    ));
                    
                }
            });
    }
}
/*
流量域 版本-渠道-地区-访客类别 粒度 页面浏览各窗口汇总表

1. 数据源
    dwd 层: 页面日志 dwd_traffic_page


2.
维度                   pv  uv
a-xiaomi-shenzhen-1   1   1
a-xiaomi-shenzhen-1   1   0
a-xiaomi-shenzhen-1   1   0
...


开窗聚合
    keyBy-> 版本-渠道-地区-访客类别
    聚合: reduce, Aggregate, process
结果:
窗口      维度                pv  uv
0-5  a-xiaomi-shenzhen-1   100   50
0-5  b-xiaomi-shenzhen-1   100   50
0-5  a-huawei-shenzhen-1   100   50
    .....
    
// 学过的聚和函数:

简单
    sum
    max min
    maxBy minBy

复杂
    reduce 增量
        如果输入的类型和聚合后的类型一致
    
    Aggregate 增量
        如果输入的类型和聚合后的类型不一致( 累加器)
    
    process 全量
        一般是业务比较复杂的时候使用.
    
 */